package org.shj.hadoop.customerinputformat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputSplit;

public class FindMaxValueInputSplit extends InputSplit implements Writable{

	private int m_StartIndex;
	private int m_EndIndex;
	private ArrayWritable m_FloatArray = new ArrayWritable(FloatWritable.class);
	
	public FindMaxValueInputSplit(){}
	
	public FindMaxValueInputSplit(int start, int end){
		m_StartIndex = start;
		m_EndIndex = end;
		int len = m_EndIndex - m_StartIndex +1;
		int index = m_StartIndex;
		FloatWritable[] result = new FloatWritable[len];
		for(int i = 0 ; i < len; i++){
			float f = FindMaxValueInputFormat.floatvalues[index];
			FloatWritable fw = new FloatWritable();
			fw.set(f);
			
			result[i] = fw;
			index++;
		}
		m_FloatArray.set(result);
	}
	
	public long getLength() throws IOException, InterruptedException{
		return this.m_EndIndex - this.m_StartIndex + 1;
	}
	
	public String[] getLocations() throws IOException, InterruptedException{
		return new String[]{"hadoop-1", "hadoop-2"};
	}

	public int getM_StartIndex() {
		return m_StartIndex;
	}

	public void setM_StartIndex(int m_StartIndex) {
		this.m_StartIndex = m_StartIndex;
	}

	public int getM_EndIndex() {
		return m_EndIndex;
	}

	public void setM_EndIndex(int m_EndIndex) {
		this.m_EndIndex = m_EndIndex;
	}

	public ArrayWritable getM_FloatArray() {
		return m_FloatArray;
	}

	public void setM_FloatArray(ArrayWritable m_FloatArray) {
		this.m_FloatArray = m_FloatArray;
	}
	
	public void readFields(DataInput in) throws IOException{
		this.m_StartIndex= in.readInt();
		this.m_EndIndex = in.readInt();
		this.m_FloatArray.readFields(in);
	}
	
	public void write(DataOutput out) throws IOException{
		out.writeInt(this.m_StartIndex);
		out.writeInt(this.m_EndIndex);
		this.m_FloatArray.write(out);
	}
}
